// -*-mode:c++; coding:utf-8-*-


#include <algorithm>
#include <limits>

#include <boost/foreach.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/function.hpp>

#include <constants.hpp>
#include <protocol.hpp>
#include <fs.hpp>

#include "global.hpp"
#include "group_scanner.hpp"
#include "naming_strategy.hpp"
#include "global.hpp"
#include "table_schema_manager.hpp"
#include "table_schema.hpp"

#include "group.hpp"

namespace {
	struct file_info_less {
		bool operator() (const fs::file_info &lhs,
				 const fs::file_info &rhs) {
			return boost::lexical_cast<uint64_t>(fs::get_name(lhs))
				<
				boost::lexical_cast<uint64_t>(fs::get_name(rhs));
		}
	};

	// 用于主要紧缩的扫描器，主要紧缩时需要分别对每个列族，按照合适
	// 的参数，打开一个Scanner，并用该类来合并它们。相当于简化的
	// tablet_scanner！
	struct major_compact_scanner
	{
		struct greater_compare {
			inline
			bool operator() (const group_scanner::pointer &first,
					 const group_scanner::pointer &second) const {
				key_less_than key_lt;
				// second.entry.key < first.entry.key => first > second
				return key_lt(*second->get().m_key,
					      *first->get().m_key);
			}
		};

		bool eos() const {
			return m_queue.empty();
		}

		const block_entry &get() const {
			assert(! m_queue.empty());
			return m_queue.top()->get();
		}

		void next() {
			if(! m_queue.empty())
			{
				group_scanner::pointer scanner = m_queue.top();
				m_queue.pop();
				scanner->next();
				const bool eos = scanner->eos();
				if(! eos)
				{
					m_queue.push(scanner);
				}
			}
		}

		void add(const group_scanner::pointer &scanner) {
			const bool eos = scanner->eos();
			if(! eos)
			{
				m_queue.push(scanner);
			}
		}
		
	private:
		std::priority_queue<group_scanner::pointer,
				    std::vector<group_scanner::pointer>,
				    greater_compare> m_queue;
	};

	// 用于合并紧缩的扫描器，合并紧缩不对数据进行过滤，直接对一部分
	// sstable获取scanner，并依次扫描即可。
	struct merge_compact_scanner
	{
		struct greater_compare {
			inline
			bool operator() (const kv_scanner::pointer &first,
					 const kv_scanner::pointer &second) const {
				key_less_than key_lt;
				// second.entry.key < first.entry.key => first > second
				return key_lt(*second->get().m_key,
					      *first->get().m_key);
			}
		};

		bool eos() const {
			return m_queue.empty();
		}

		const block_entry &get() const {
			assert(! m_queue.empty());
			return m_queue.top()->get();
		}

		void next() {
			if(! m_queue.empty())
			{
				kv_scanner::pointer scanner = m_queue.top();
				m_queue.pop();
				scanner->next();
				const bool eos = scanner->eos();
				if(! eos)
				{
					m_queue.push(scanner);
				}
			}
		}

		void add(const kv_scanner::pointer &scanner) {
			const bool eos = scanner->eos();
			if(! eos)
			{
				m_queue.push(scanner);
			}
		}
		
	private:
		std::priority_queue<kv_scanner::pointer,
				    std::vector<kv_scanner::pointer>,
				    greater_compare> m_queue;
	};

}

namespace xbase {

	group::pointer group::load (const std::string &p_tablet_path,
				    const std::string &p_table_name,
				    const std::string &p_group_name,
				    const uint64_t &tablet_id,
				    const std::string &p_tablet_name,
				    const std::string &start_row,
				    const std::string &end_row,
				    const table_schema::group_info &info,
				    const family_map_type &families)
	{
		// first, group path is ok
		const std::string _group_path = naming_strategy::group_path(p_tablet_path, p_group_name);
		const fs::path _path = _group_path;
		
		group::pointer _group(new group(p_table_name,
						p_group_name,
						tablet_id,
						p_tablet_name,
						_path,
						start_row,
						end_row,
						info,
						families));

		// try {

		std::vector<fs::file_info> _file_list;
		const bool ok = fs::list_files(_file_list, _group_path);
		if(! ok)
		{
			LOG(ERROR, "group::load list directory " << _group_path
			    << " failed, fs::errno = "
			    << fs::get_errno());
			return group::pointer();
		}

		// sort files by type
		std::vector<fs::file_info> _sstable_file_list;
		std::vector<fs::file_info> _sstable_ref_file_list;
		BOOST_FOREACH(const fs::file_info &_info, _file_list)
		{
			const std::string _name = fs::get_name(_info);
			if (naming_strategy::is_sstable_file(_name))
			{
				_sstable_file_list.push_back(_info);
			}
			else if (naming_strategy::is_sstable_ref(_name))
			{
				_sstable_ref_file_list.push_back(_info);
			}
			else	// other
			{
				if (naming_strategy::is_trash_file(fs::get_name(_info))
				    || naming_strategy::is_deleted_file(fs::get_name(_info)))
				{
					const std::string from_path =
						(_path / fs::get_name(_info)).string();
					std::string dest_name = from_path;
					const std::string root_path = global::xbase_root_path().string();
					if((dest_name.size() > root_path.size()) &&
					   (dest_name.compare(0, root_path.size(), root_path) == 0))
					{
						// 将前缀去掉
						if(dest_name[root_path.size()] == '/')
							dest_name.erase(0, root_path.size() + 1); // 这是正常情况
						else
							dest_name.erase(0, root_path.size());
					}
					std::replace(dest_name.begin(), dest_name.end(), '/', '_');
					const fs::path dest_path =
						global::xbase_root_path() / ".trash" / dest_name;
					LOG_ENV(const bool ok_rename = ) fs::rename(from_path, dest_path);
					LOG(WARN, "found trash/deleted file " << from_path
					    << ", move it to .trash: " << dest_path);
					LOG_IF((! ok_rename), WARN,
					       "buf rename failed, fs::errno = " << fs::get_errno());
				}
				else
				{
					LOG(WARN, "group::load found unknown file "
					    << fs::get_name(_info) << ", trash it");
					fs::rename((_path / fs::get_name(_info)),
						   (_path / naming_strategy::trash_filename(fs::get_name(_info))));
				}
			}
		}

		// 按顺序加载SSTable，首先加载sstable_ref（因为它肯定是
		// 最早的），其它SSTable排序后依次加载。

		if (_sstable_ref_file_list.size() == 1)
		{
			const std::string &name = fs::get_name(_sstable_ref_file_list.front());
			// parse ref sstable info
			const std::string ref_tablet_pathname = naming_strategy::get_tablet_pathname_from_sstable_ref_name(name);
			fs::path ref_path = naming_strategy::ref_tablet_path(global::xbase_root_path().string(),
									     ref_tablet_pathname);
			ref_path /= naming_strategy::group_dirname(p_group_name); // under the same group dir

			std::vector<fs::file_info> _list; // find ref sstable
			const bool _ok = fs::list_files(_list, ref_path);
			if(! _ok)
			{
				LOG(ERROR,"group::load list directory "
				    << ref_path << " error, fs::errno = "
				    << fs::get_errno());
				return group::pointer();
			}

			bool _found = false;
			std::string _sstable_filename;
			BOOST_FOREACH(const fs::file_info &_info, _list)
			{
				if (naming_strategy::is_sstable_file(fs::get_name(_info))) // found sstable file
				{
					if (! _found) // firstly found
					{
						if (fs::is_regular(_info))
						{
							_sstable_filename = fs::get_name(_info);
							_found = true;
						}
						else
						{
							LOG(WARN,"group::load ref'ed sstable file "<<fs::get_name(_info)
							    <<" in ref path " << ref_path
							    << " is not a regular file, skip it");
							// fs::remove(ref_path / fs::get_name(_info));
						}
					}
					else // found multi-sstable file
					{ 
						LOG(ERROR, "group::load found more than 1 sstable file in ref'ed tablet dir "
						    << ref_path);
						return group::pointer();
					}
				}
				else // other file, clean up
				{
					if(naming_strategy::is_trash_file(fs::get_name(_info))
					   || naming_strategy::is_deleted_file(fs::get_name(_info)))
					{
						const std::string from_path =
							(ref_path / fs::get_name(_info)).string();
						std::string dest_name = from_path;
						const std::string root_path = global::xbase_root_path().string();
						if((dest_name.size() > root_path.size()) &&
						   (dest_name.compare(0, root_path.size(), root_path) == 0))
						{
							// 将前缀去掉
							if(dest_name[root_path.size()] == '/')
								dest_name.erase(0, root_path.size() + 1); // 这是正常情况
							else
								dest_name.erase(0, root_path.size());
						}
						std::replace(dest_name.begin(), dest_name.end(), '/', '_');
						const fs::path dest_path =
							global::xbase_root_path() / ".trash" / dest_name;
						LOG_ENV(const bool ok_rename = ) fs::rename(from_path, dest_path);
						LOG(WARN, "found trash/deleted file " << fs::get_name(_info)
						    << " in parent tablet path " << ref_path
						    << ", move it to .trash: " << dest_path);
						LOG_IF((! ok_rename), WARN,
						       "but rename failed, fs::errno = " << fs::get_errno());
					}
					else
					{
						//fs::remove(ref_path / fs::get_name(_info)); // do not need to check return value
						LOG(WARN,"group::load found unknown file "<<fs::get_name(_info)
						    << " in ref path " << ref_path
						    <<", trash it");
						fs::rename((ref_path / fs::get_name(_info)),
							   (ref_path / naming_strategy::trash_filename(fs::get_name(_info))));
					}
				}
			}
			if(! _found) // not found
			{
				LOG(ERROR, "group::load ref sstable file " << name << " in group path"
				    << _path << " lost target file " << ref_path);
				return group::pointer();
			}
			ref_path /= _sstable_filename; // ref'ed sstable path
			
			// load ref sstable
			const uint64_t seq = naming_strategy::get_timestamp_from_sstable_name(name);
			const sstable_type type = ref_to_which(tablet_id);
			LOG(DEBUG, "group::load sstable " << name
			    << " in " << _path << " is ref to "
			    << ((type == ST_REF_TOP_SSTABLE) ? "top half" : "bottom half")
			    << " of parent sstable " << ref_path);

			sstable::pointer _sstable = sstable::load(ref_path.string(),
								  (_path / name).string(),
								  (_path / naming_strategy::deleted_filename(name)).string(),
								  start_row,
								  end_row,
								  seq,
								  type);
			if (! _sstable)
			{
				LOG(ERROR, "group::load loading ref sstable "
				    << name << " in path " << _path
				    << " failed, real ref_path = " << ref_path);
				return group::pointer();
				// fs::rename((_path / fs::get_name(_sstable_ref_file_list.front())),
				// 	   (_path / naming_strategy::trash_filename(fs::get_name(_sstable_ref_file_list.front()))));
			}
			else
			{
				_group->append(_sstable);
				LOG(DEBUG, "group::load ref sstable "
				    << name
				    << " loaded.");
			}
		}
		else if (_sstable_ref_file_list.size() > 1)
		{
			LOG_ENV(std::ostringstream out;
				for(std::size_t i=0; i<_sstable_ref_file_list.size(); ++i)
				out << "\t"
				    << fs::get_name(_sstable_ref_file_list[i])
				    << "\n");
			LOG(ERROR, "group::load found more than one ref sstable in group path " << _path
			    << ", list:\n" << out.str());
			return group::pointer();
		}
		// else, is empty, do nothing

		// 对文件进行排序，使其按时间顺序加载
		std::sort(_sstable_file_list.begin(),
			  _sstable_file_list.end(),
			  file_info_less());
		BOOST_FOREACH(const fs::file_info &_info, _sstable_file_list)
		{
			const std::string &name = fs::get_name(_info);

			// 在SSTable内部加载时，再根据文件类型进行适当的加载
			sstable::pointer _sstable = sstable::load((_path / name).string(),
								  (_path / name).string(),
						    		  (_path / naming_strategy::deleted_filename(name)).string(),
								  start_row,
								  end_row,
								  naming_strategy::get_timestamp_from_sstable_name(name),
								  ST_NORMAL_SSTABLE);
			if (! _sstable)
			{
				LOG(FATAL, "group::load loading sstable "
				    << name << " in group path " << _path
				    << " failed, real path = " << (_path / name));
				return group::pointer();
				// fs::rename((_path / fs::get_name(_info)),
				// 	   (_path / naming_strategy::trash_filename(fs::get_name(_info))));
			}
			else
			{
				_group->append(_sstable);
				LOG(DEBUG, "group::load sstable "
				    << name << " in group path " << _path
				    << " loaded");
			}
		}

		// 初始设置合并紧缩计数为加载后的SSTable个数
		_group->m_merge_counter = _group->m_sstables.size();

		return _group;
	}

	group::group(const std::string &p_table_name,
		     const std::string &p_group_name,
		     const uint64_t &tablet_id,
		     const std::string &p_tablet_name,
		     const fs::path &p_path,
		     const std::string &start_row,
		     const std::string &end_row,
		     const table_schema::group_info &info,
		     const family_map_type &families)
		: m_table_name(p_table_name)
		, m_group_name(p_group_name)
		, m_tablet_id(tablet_id)
		, m_tablet_name(p_tablet_name)
		, m_path(p_path)
		, m_start_row(start_row)
		, m_end_row(end_row)
		, m_group_info(info)
		, m_family_list(families)
		, m_lower_bound_of_max_num_sstable(static_cast<std::size_t>(global::the_conf->lower_bound_of_max_num_sstable))
		, m_max_sequence_number(0)
		, m_merge_counter(0) {
		m_memtable.reset(new memory_table(info.m_max_size_memtable,
						  global::the_conf->minor_compact_interval,
						  p_table_name,
						  p_group_name,
						  tablet_id,
						  p_tablet_name,
						  start_row,
						  end_row,
						  m_group_info,
						  m_family_list));
	}
	
	void group::append (sstable::pointer p_sstable)
	{
		//Guard<ReadWriteLock::WriteLock> _guard(m_sstables_lock.write_lock());
		boost::unique_lock<boost::shared_mutex> _guard(m_sstables_lock);
		m_sstables.push_back(p_sstable);
		m_max_sequence_number = (std::max)(m_max_sequence_number, p_sstable->sequence_number());
	}

	// DONE:潜在的问题,已经解决（通过锁定，在获取Scanner的时候不让
	// 内存表和SSTable发生变化）
	// 
	// 如果在一个scanner周期中，内存表发生了紧缩，从而使SSTable增
	// 加，而此时原来的Scanner中就找不到该SSTable中的数据了
	//
	// 另外，若组发生了主要紧缩，所有的SSTable都不存在的，Scanner就
	// 会出错，可能会使客户端从头扫描。
	//
	// 如果能在组内SSTable数量变化的时候（增加或减少），通知该组的
	// 已有的Scanner，来相应的更新其内部组成，应该可以解决这个问题。
	// （但涉及到扫描位置重新定位的过程）
	//
	// 对子表来说，若组可能会发生变动的话，也是有同样的问题的。
	group_scanner::pointer group::get_scanner (kv_scan_context::pointer p_context)
	{
		group_scanner::pointer _scanner(new group_scanner(p_context));
		if(! _scanner)
		{
			LOG(ERROR, "group::get_scanner new group_scanner failed, maybe memory not enough");
			return group_scanner::pointer();
		}

		// 重要！
		// 通过对内存表的历史表加读锁，阻止在获取扫描器时删除历史表及创建SSTable，否则可能会导致扫描的对象缺少或重复
		//
		// 不能通过对组的SSTable队列加读锁实现，这样虽然不能加
		// 入新SSTable，但不能阻止历史表删除，会导致扫描的对象
		// 缺少（比如一个历史表刚好次要紧缩完并删除，但又没加入到SSTable队列中）。
		//
		// 必须在整个函数范围内加锁
		//Guard<ReadWriteLock::ReadLock> guard(m_memtable->history_table_lock().read_lock());
		boost::shared_lock<boost::shared_mutex> guard(m_memtable->history_table_lock());

		kv_scanner::pointer _p = m_memtable->get_scanner(p_context);
		if (! _p)
		{
			LOG(TRACE, "group::get_scanner null memory table scanner, skip it");
		}
		else
		{
			_scanner->add(_p);
		}

		// 这里添加SSTable的顺序已经不重要，即sstables中的
		// SSTable顺序已经不重要，因为在扫描器中已经考虑到了
		// SSTable的顺序号（文件名）。
		//
		// 但以后实现读操作时，也许还会依赖于这里的顺序。
		{
			//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
			BOOST_FOREACH(const sstable::pointer &_sstable, m_sstables)
			{
				kv_scanner::pointer __p = _sstable->get_scanner(p_context);
				if (! __p)
				{
					LOG(TRACE, "group::get_scanner null sstable scanner in "
					    << _sstable->path() << ", skip it");
				}
				else
				{
					_scanner->add(__p);
				}
			}
		}

		// 重要，添加完成后需要调用INIT函数
		_scanner->init();

		return _scanner;
	}

	group_scanner::pointer group::get_sstables_scanner (kv_scan_context::pointer p_context)
	{
		group_scanner::pointer _scanner(new group_scanner(p_context));

		//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
		BOOST_FOREACH(const sstable::pointer &_sstable, m_sstables)
		{
			kv_scanner::pointer __p = _sstable->get_scanner(p_context);
			if (! __p)
			{
				LOG(WARN, "group::get_scanner null sstable scanner in "
				    << _sstable->path() << ", skip it");
			}
			else
			{
				_scanner->add(__p);
			}
		}

		// 重要，添加完成后需要调用INIT函数
		_scanner->init();

		return _scanner;
	}

	group_scanner::pointer group::get_sstables_scanner_no_lock (kv_scan_context::pointer p_context)
	{
		group_scanner::pointer _scanner(new group_scanner(p_context));

		//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
		//boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
		BOOST_FOREACH(const sstable::pointer &_sstable, m_sstables)
		{
			kv_scanner::pointer __p = _sstable->get_scanner(p_context);
			if (! __p)
			{
				LOG(WARN, "group::get_scanner null sstable scanner in "
				    << _sstable->path() << ", skip it");
			}
			else
			{
				_scanner->add(__p);
			}
		}

		// 重要，添加完成后需要调用INIT函数
		_scanner->init();

		return _scanner;
	}

	std::string group::to_string () const
	{
		return group_name();
	}

	// recover mode minor compact
	bool group::recover_compact()
	{
		if(m_memtable->empty())
		{
			return true;
		}
		
		uint64_t _seq_num = this->max_sequence_number() + 1;
		if(_seq_num == 1) // 如果没有SSTable，则用当前时间或当前的序列号来命名文件（都可以）。
		{
			_seq_num = cur_micro_seconds();//global::the_log_manager->logger().get_sequence_number();
		}
		LOG(DEBUG, "group::recover_compact compacting with sequence number "
		    << _seq_num
		    << " in group " << group_name()
		    << " of tablet " << tablet_name() << ".");
		sstable::pointer _result = m_memtable->compact(_seq_num);
		if (! _result)
		{
			LOG(INFO, "group::recover_compact all data recoved has out of date");
		}
		else
		{
			this->append(_result);
		}
		return true;
	}

	bool group::minor_compact(bool p_forced)
	{
		if (m_memtable->empty() || // 内存表为空，无条件返回
		    ((! p_forced) && (! need_minor_compact())))
		{
			LOG(DEBUG, "group::minor_compact minor compacting not needed yet.");
			return true;
		}

		sstable::pointer _result;
		{
			boost::mutex::scoped_lock guard(global::the_log_manager->logger().roll_back_lock());
			const uint64_t _seq_num = global::the_log_manager->logger().get_sequence_number();
			LOG(DEBUG, "group::minor_compact compacting with sequence number "
			    << _seq_num
			    << " in group " << group_name()
			    << " of tablet " << tablet_name());
			_result = m_memtable->compact(_seq_num);
			global::the_log_manager->logger().
				complete_minor_compact(naming_strategy::composited_tablet_group_name(tablet_name(),
												     group_name()),
						       _seq_num);
		}
		if (! _result)
		{
			LOG(WARN, "group::minor_compact minor compact return null sstable");
			// ignore it,因为数据可能全部失效了。而如果是写
			// 文件出错的话，在紧缩过程中就会退出程序。
		}
		else
		{
			this->append(_result);
		}
		return true;
	}

	// 在内存表、所有SSTable上用合适的参数（过滤无用的数据）打开
	// scanner，然后写入到新的SSTable(用最近的一个SSTable的文件名)
	// 中。
	//
	// 记得要通知log线程，完成了紧缩
	//
	// 要保证主要紧缩完成后，内存表中没有任何数据，有且仅有一个
	// SSTable
	void group::major_compact()
	{
		// 若应进行合并紧缩，则转到合并紧缩。
		// 否则，主要紧缩完成后，则应置m_merge_counter为SSTable个数
		if(should_merge_compact())
		{
			merge_compact();
			return;
		}

		LOG(TRACE, "group::major_compact starting major compact.");
		LOG_ENV(time_t start_time = cur_seconds());

		// 0. must lock the tablet for write (and read?)
		//
		// do it in tablet level

		// 1. do a minor compact, get cur seq num
		//
		// the max_sequence_number updated
		// this->minor_compact(true);
		// 不需要这个次要紧缩了，因为现在的主要紧缩就是由次要紧缩触发的

		// 这里不能说只有一个SSTable的时候就不进行紧缩，因为它可能是引用SSTable，因此需要进行判断。
		{
			//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
			if (m_sstables.size() == 0) { // no sstable, do nothing
				LOG(DEBUG, "group::major_compact"
				    << " no sstable file, do nothing.");
				return;
			}
			////TODO:? 即使只有一个SSTable，还是继续主要紧缩，因为里面可能含有一些无效的数据
			else if (m_sstables.size() == 1) { // only one sstable
				if (! m_sstables.front()->is_ref_sstable()) { // and not a ref sstable, do nothing
					LOG(DEBUG, "group::major_compact"
					    << " only one regular sstable file, so do nothing.");
					return;
				} else {
					LOG(DEBUG, "group::major_compact"
					    << " one ref sstable file, do major compact.");
				}
			}
		}

		// 
		// 2. create scan context
		//

		// 
		// 3. get a scanner from this group
		//
		major_compact_scanner _scanner;
		typedef std::vector<sstable::pointer>::iterator it_t;
		std::size_t major_compact_number = 0; // 保存主要合并sstable数
		// construct scanner by family
		{
			// 整个过程中，sstables队列不能改变，因此在这里加锁
			boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
			major_compact_number = m_sstables.size(); // save
			
			std::vector<std::string> columns(1); // 一个元素即可

			BOOST_FOREACH(const family_map_type::value_type &elem, m_family_list)
			{
				columns[0] = elem.first + ":*"; // IMPORTANT: only cur family
				// 根据dead_time得出最老时间戳
				const uint64_t oldest_ts = (elem.second.m_dead_time == 0
							    ? NO_TIMESTAMP
							    : uint64_t(elem.second.m_dead_time) * 1000ULL * 1000ULL); 
				kv_scan_context::pointer context(new kv_scan_context(this->tablet_name(),
										     m_start_row, // IMPORTANT: tablet start row
										     m_end_row, // IMPORTANT: tablet end row
										     columns,
										     NO_TIMESTAMP, // latest_ts,
										     oldest_ts,
										     elem.second.m_max_ttl,
										     elem.second.m_max_num_version, // IMPORTANT: !
										     (kv_scan_context::OPTION_DISABLE_FILTER_DELETED_ITSELF), // options: filter deleted and not filter deletion itself
										     1)); // num row
				group_scanner::pointer scanner_for_family = this->get_sstables_scanner_no_lock(context);
				if (! scanner_for_family)
				{
					LOG(ERROR, "group::major_compact"
					    << " get scanner for family "
					    << elem.first
					    << " in group "
					    << this->group_name()
					    << " failed");
					return;
				}
				_scanner.add(scanner_for_family);
				LOG(DEBUG, "group::major_compact got scanner for family "
				    << elem.first << " in group " << this->group_name());
			}
		}

		// 
		// 5. read from composited-scanner and append into new
		// sstable
		sstable::pointer _new_sstable;
		if(_scanner.eos())
		{
			LOG(DEBUG, "scanner is empty, can delete all sstables");
			// 这里不要返回，要到后面的处理逻辑
		}
		else
		{
			// // 取当前的日志序列号，作为新SSTable的文件名；因为此时该组的日志一定已经全部应用，并且没有残留。
			// // 并且这样取的文件名，不和已有的文件名重复，方便后续操作。
			// const uint64_t seq_num = global::the_log_manager->logger().get_sequence_number();
			// 
			// 因为现在主要紧缩过程中仍然允许写记录，所以这里不能取当前日志号作为新表名了。因此采用最大序列号加1
			const uint64_t seq_num = this->max_sequence_number() + 1;
			assert(seq_num > 1);
			LOG(INFO, "group::major_compact compacting with sequence number "
			    << seq_num << " in group " << group_name()
			    << " of tablet " << tablet_name());
			_new_sstable = sstable::create(naming_strategy::sstable_path(m_path.string(),
										     seq_num),
						       naming_strategy::compact_sstable_path(m_path.string(),
											     seq_num),
						       naming_strategy::deleted_sstable_path(m_path.string(),
											     seq_num),
						       m_start_row,
						       m_end_row,
						       seq_num,
						       m_group_info.m_block_size,
						       static_cast<compressor::compressor_code>(m_group_info.m_compress_method));
			if(! _new_sstable)
			{
				LOG(ERROR, "group::major_compact create sstable with seq_num "
				    << seq_num
				    << " in group path " << m_path
				    << " failed, abort");
				return;
			}
			
			LOG_ENV(int kv_count = 0);
			for(; ! _scanner.eos(); _scanner.next())
			{
				const bool _ok = _new_sstable->append(_scanner.get());
				if (! _ok)
				{
					LOG(ERROR, "group::major_compact"
					    << " appending record ("
					    << dump_string_char(*_scanner.get().m_key->row())
					    << ", "
					    << dump_string_char(*_scanner.get().m_key->column())
					    << ", "
					    << _scanner.get().m_key->timestamp()->get()
					    << ") into new sstable failed.");
					return;
				}
				LOG_ENV(++kv_count);
			}
			
			LOG(DEBUG, "group::major_compact "
			    << kv_count << " records written to sstable "
			    << _new_sstable->path());

			// and then rename the new sstable file
			// 因为文件名不重复，所以可以先生成新的SSTable，再删除旧的。
			const bool ok = _new_sstable->append_eof();
			if(! ok)
			{
				LOG(ERROR, "group::major_compact append eof failed, "
				    << "removing the new sstable file " << _new_sstable->path()
				    << " and aborting major compact");
				_new_sstable->set_deleted(); // 通过path()得到的不一定是合适的文件名
				// LOG_ENV(const bool ok = )fs::remove(m_path / _new_sstable->path());
				// LOG_IF((!ok), ERROR,
				// 	"group::major_compact remove file failed, fs::errno = "
				//        << fs::get_errno() << ", ignore it");
				return;
			}
		}

		// 
		// 6. remove all old sstable
		// clear old sstable and sstable list
		{
			//Guard<ReadWriteLock::WriteLock> _guard(m_sstables_lock.write_lock());
			boost::unique_lock<boost::shared_mutex> _guard(m_sstables_lock);
			//BOOST_FOREACH(sstable::pointer &__sstable, m_sstables)
			const it_t range_begin = m_sstables.begin();
			const it_t range_end = range_begin + major_compact_number;
			assert(range_end <= m_sstables.end());
			for(it_t it = range_begin; it != range_end; ++it)
			{
				sstable::pointer &__sstable = *it;
					
				// 设置删除状态，令其在析构时被删除，这
				// 里会先将SSTable改名为不合法的文件
				// 名，这样程序core掉后，再次加载也不会
				// 导致该数据继续保留下来。
				LOG_ENV(const bool __ok = )__sstable->set_deleted();
				LOG_IF_ELSE((! __ok),
					    ERROR,
					    "group::major_compact set_deleted for old sstable file "
					    << __sstable->path() << " failed, ignore it",
					    INFO,
					    "group::major_compact old sstable file "
					    << __sstable->path() << " set deleted");
				// 如果这里删除旧文件出错，则会导致数据
				// 重复。但对数据影响不大，下次紧缩时，
				// 还会再次删除掉它。
				//
				// 这就要求，在读取数据的时候，要严格按
				// 照SSTable文件的新旧顺序来读，如果
				// Key完全相同，则应该取较新的文件来读！
			}

			// clear
			/*const it_t insert_pos = */this->m_sstables.erase(range_begin, range_end);
			
			// >>> 注意 <<<
			// 
			// 检查主要紧缩过程中是否有新sstable加入，现在
			// 假设是没有（即主要紧缩期间不能进行次要紧
			// 缩）；但若这个条件不满足，则需要更改检查分裂
			// 的逻辑，或者修改实现使其符合该要求（这样最好）
			// LOG_IF((! ((insert_pos == m_sstables.begin()) &&
			// 	   (insert_pos == m_sstables.end()))),
			LOG_IF((! m_sstables.empty()),
			       FATAL,
			       ">>> ERROR: minor compacting occurs during major compacting, FIX ME! <<<");
			assert(m_sstables.empty());

			// add new sstable
			// 主要紧缩后，重置合并紧缩计数为当前大SSTable个数，实际情况为1（或0）
			if(_new_sstable)
			{
				this->m_sstables.push_back(_new_sstable);
				m_max_sequence_number = (std::max)(m_max_sequence_number, _new_sstable->sequence_number());
				m_merge_counter = 1;
			}
			else
			{
				m_merge_counter = 0;
			}
		}
		
		LOG(INFO, "group::major_compact used "
		    << cur_seconds() - start_time << " seconds");
	}

	bool group::is_need_split() const {
		// assert(m_sstables.size() <= 1)
		bool _can_split = false;
		//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
		// 当且仅当SSTable数为1时，且不是引用SSTable时，才允许
		// 进行分裂
		// 
		// 若有引用SSTable，不允许分裂！但若强制分裂的话，需要
		// 进行强制紧缩，此时会将该引用SSTable变为非引用的，从
		// 而可以分裂。
		//
		_can_split = ((m_sstables.size() == 1)
			      && (m_sstables.front()->is_ref_sstable() == false));

		bool _need_split = false;
		if (_can_split)
		{
			const uint64_t _file_size = this->total_file_size_no_lock();
			_need_split = (_file_size >= m_group_info.m_max_size_sstable);
			LOG(DEBUG, "group::is_need_split"
			    << (_need_split ? " need " : " not need ")
			    << " to split the group "
			    << this->group_name() << " with filesize "
			    << _file_size << " (max size "
			    << m_group_info.m_max_size_sstable << ").");
		} // else keep _need_split is false

		return _need_split;
	}

	uint64_t group::total_file_size_no_lock() const
	{
		uint64_t _size = 0;
		//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
		//boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
		BOOST_FOREACH(const sstable::pointer &_sstable, m_sstables)
		{
			_size += _sstable->file_size();
		}
		return _size;
	}

	bool group::need_major_compact() const
	{
		const std::size_t max_num_in_group = static_cast<std::size_t>(m_group_info.m_max_num_sstable);

		// SSTable数量的下限
		const std::size_t _max = (std::max)(m_lower_bound_of_max_num_sstable,
						    max_num_in_group);
		//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
		if(this->m_sstables.size() >= _max)
		{
			return true;
		}
		else
		{
			// 如果文件数不够，但文件总大小达到一定值，则需要主要紧缩
			const uint64_t file_size = total_file_size_no_lock();
			const uint64_t max_size = data_size_threshold();
			return (file_size >= max_size);
		}
	}

	bool group::split(const std::string &p_midrow,
			  const std::string &p_path_a,
			  const uint64_t p_tablet_a_id,
			  const std::string &p_tablet_a_name,
			  const std::string &p_path_b,
			  const uint64_t p_tablet_b_id,
			  const std::string &p_tablet_b_name,
			  const uint64_t p_sstable_sequence_number)
	{
		// assert only ONE or ZERO sstable!

		const std::string group_path_a = naming_strategy::group_path(p_path_a, this->group_name());
		const std::string group_path_b = naming_strategy::group_path(p_path_b, this->group_name());
		// 1. create group dir
		bool _ok = fs::mkdir(group_path_a);
		if (! _ok) {
			LOG(ERROR, "create group a " << group_path_a
			    << " failed, fs::errno = "
			    << fs::get_errno());
			return false;
		}
		_ok = fs::mkdir(group_path_b);
		if (! _ok) {
			LOG(ERROR, "create group b " << group_path_b
			    << " failed, fs::errno = "
			    << fs::get_errno());
			
			fs::remove(group_path_a); // clean up
			return false;
		}

		// 2. create ref sstable file, if has sstable
		{
			//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
			if (m_sstables.empty()) {
				// do not create ref sstable file
				LOG(DEBUG, "group::split no sstable to be referenced to.");
			} else if (m_sstables.size() == 1) {
				// create ref sstable file
				const std::string sstable_ref_a = naming_strategy::sstable_ref_path(group_path_a,
												    p_sstable_sequence_number,
												    m_table_name,
												    m_tablet_id,
												    m_start_row,
												    m_end_row);
				const std::string sstable_ref_b = naming_strategy::sstable_ref_path(group_path_b,
												    p_sstable_sequence_number,
												    m_table_name,
												    m_tablet_id,
												    m_start_row,
												    m_end_row);
				const fs::file_t _file_a = fs::create(sstable_ref_a);
				_ok = (_file_a != fs::BAD_FILE);
				if(_ok)
				{
					fs::close(_file_a);
				}
				else
				{
					LOG(ERROR, "create sstable ref a " << sstable_ref_a
					    << " failed, fs::errno = " << fs::get_errno());
					
					fs::remove(group_path_a); // clean up
					fs::remove(group_path_b);
					return false;
				}
				const fs::file_t _file_b = fs::create(sstable_ref_b);
				_ok = (_file_b != fs::BAD_FILE);
				if(_ok)
				{
					fs::close(_file_b);
				}
				else
				{
					LOG(ERROR, "create sstable ref b " << sstable_ref_b
					    << " failed, fs::errno = "
					    << fs::get_errno());
					
					fs::remove(sstable_ref_a);
					fs::remove(group_path_a); // clean up
					fs::remove(group_path_b);
					return false;
				}
			} else {
				// error, more than one sstable files
				LOG(ERROR, "group::split more than one sstable files, abort.");
				fs::remove(group_path_a); // clean up
				fs::remove(group_path_b);
				return false;
			}
		}

		// 3. if memtable not empty, compact it to child tablets path
		if(! m_memtable->empty())
		{
			boost::mutex::scoped_lock guard(global::the_log_manager->logger().roll_back_lock());
			const uint64_t _seq_num = global::the_log_manager->logger().get_sequence_number();
			LOG(DEBUG, "group::split compacting with sequence number "
			    << _seq_num
			    << " in group " << group_name()
			    << " of tablet " << tablet_name());

			LOG_ENV(int counter_a = 0);
			LOG_ENV(int counter_b = 0);
			m_memtable->freeze();
			minor_compact_scanner::pointer scanner = m_memtable->get_compact_scanner();
			if(! scanner)
			{
				LOG(ERROR, "group::split get compact scanner failed");
				return false;
			}
			
			sstable::pointer sstable_a, sstable_b;
			for(; ! scanner->eos(); scanner->next())
			{
				const block_entry &entry = scanner->get();
				if(*(entry.m_key->row()) < p_midrow) // 属于上半个子表
				{
					if(! sstable_a)
					{
						sstable_a = sstable::create(naming_strategy::sstable_path(group_path_a,
													  _seq_num),
									    naming_strategy::compact_sstable_path(group_path_a,
														  _seq_num),
									    naming_strategy::deleted_sstable_path(group_path_a,
														  _seq_num),
									    m_start_row,
									    p_midrow,
									    _seq_num,
									    m_group_info.m_block_size,
									    static_cast<compressor::compressor_code>(m_group_info.m_compress_method));
						if(! sstable_a)
						{
							LOG(ERROR, "group::split create sstable for child-tablet-a "
							    << p_tablet_a_name << " failed, abort");
							return false;
						}
					}
					const bool ok = sstable_a->append(entry);
					LOG_ENV(++counter_a);
					if(! ok)
					{
						LOG(ERROR, "group::split append " << counter_a
						    << "th record to tablet_a failed");
						return false;
					}
				}
				else
				{
					if(! sstable_b)
					{
						sstable_b = sstable::create(naming_strategy::sstable_path(group_path_b,
													  _seq_num),
									    naming_strategy::compact_sstable_path(group_path_b,
														  _seq_num),
									    naming_strategy::deleted_sstable_path(group_path_b,
														  _seq_num),
									    p_midrow,
									    m_end_row,
									    _seq_num,
									    m_group_info.m_block_size,
									    static_cast<compressor::compressor_code>(m_group_info.m_compress_method));
						if(! sstable_b)
						{
							LOG(ERROR, "group::split create sstable for child-tablet-b "
							    << p_tablet_b_name << " failed, abort");
							return false;
						}
					}
					const bool ok = sstable_b->append(entry);
					LOG_ENV(++counter_b);
					if(! ok)
					{
						LOG(ERROR, "group::split append " << counter_b
						    << "th record to tablet_b failed");
						return false;
					}
				}
			}

			if(sstable_a)
			{
				const bool ok = sstable_a->append_eof();
				if(! ok)
				{
					LOG(ERROR, "group::split append_eof on tablet_a failed");
					return false;
				}
			}
			if(sstable_b)
			{
				const bool ok = sstable_b->append_eof();
				if(! ok)
				{
					LOG(ERROR, "group::split append_eof on tablet_b failed");
					return false;
				}
			}

			// 写成功了，才可以将内存表删除，并将日志删除；否则，留下的内存表在下次次要紧缩时会再将写到磁盘上
			m_memtable->melt(); // to un-freeze
			global::the_log_manager->logger().
				complete_minor_compact(naming_strategy::composited_tablet_group_name(tablet_name(),
												     group_name()),
						       _seq_num);
		}

		LOG(DEBUG, "group::split group " << this->group_name() << " is split successfully.");
		return true;
	}

	std::string group::get_mid_row() const
	{
		std::string _mid_row;
		//Guard<ReadWriteLock::ReadLock> _guard(m_sstables_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> _guard(m_sstables_lock);
		if (m_sstables.empty()) {
			LOG(ERROR, "group::get_mid_row no sstable, return empty string.");
		} else if (m_sstables.size() == 1) {
			_mid_row = m_sstables.front()->get_split_row();
			LOG(DEBUG, "group::get_mid_row got " << _mid_row
			    << " in sstable " << m_sstables.front()->path() << ".");
		} else {
			LOG(ERROR, "group::get_mid_row more than one sstable,"
			    << " should not go here now[CHECK IT]");
		}
		return _mid_row;
	}

	group::~group()
	{
		// 必须首先删除所有的SSTable，因为SSTable析构的
		// 时候，要求该组的缓存管理单元存在！
		m_sstables.clear();
		
		// 必须删除日志中的所有日志信息，令组的日志文件可以被删除
		global::the_log_manager->logger().complete_minor_compact(naming_strategy::composited_tablet_group_name(tablet_name(),
														       group_name()),
									 std::numeric_limits<uint64_t>::max()); // 用最大值即可使信息删除
	}

	// 检查是否应该合并紧缩。仅在主要紧缩条件满足时执行该检查。检查
	// 的因素有：
	// 
	// － 合并计数：使每次合并新的紧缩紧缩生成的sstable（即较小的
	// sstable），若小sstable没有或太少，就进行主要紧缩。
	// 
	// － 总数据量：若总数据量已经足够大了，则不进行合并紧缩，直接
	// 进行主要紧缩。
	//
	// 返回值：
	//     真表示应该进行合并紧缩；否则表示应该进行主要紧缩。
	bool group::should_merge_compact() const {
		boost::shared_lock<boost::shared_mutex> lock(m_sstables_lock);
		const std::size_t sstable_number = this->m_sstables.size();
		
		const std::size_t max_sstable_num = m_group_info.m_max_num_sstable;
		const std::size_t lower_bound = m_lower_bound_of_max_num_sstable;
		const std::size_t max_value = (std::max)(max_sstable_num, lower_bound);

		// 若还没达到 major_compact 的文件数，则说明这一次是强
		// 制主要紧缩，因此不应该进行合并紧缩，应转为主要紧缩。
		//
		// 需要注意，与判断是否需要主要紧缩
		// (need_major_compact())一致。
		//
		// 这不是最好的解决方式，更好的是：分裂时（只在此时和恢
		// 复时需要强制主要紧缩）传参数给major_compact以说明来
		// 意。这点留作以后完善。
		if(sstable_number < max_value)
		{
			return false;
		}

		const std::size_t threshold = max_value - lower_bound;

		LOG_IF((max_sstable_num < m_merge_counter),
		       WARN,
		       "max_sstable_num should greater than or equal to merge_counter, max_sstable_num = " << max_sstable_num
		       << ", merge_counter = " << m_merge_counter);
		const std::size_t sstable_num_to_merge = ((max_sstable_num >= m_merge_counter)
							  ? (max_sstable_num - m_merge_counter)
							  : 0);

		// 合并紧缩计数检测，发现适合合并，则再检查文件总数据量
		if((m_merge_counter < threshold) &&
		   (sstable_num_to_merge >= lower_bound))
		{
			// 如果文件总数据足够多了，则不进行合并紧缩，转入主要紧
			// 缩。考虑到主要紧缩后数据可能减少，足够多定义为，当前
			// 大小超过最大大小25%以上，这样主要紧缩后，通常都能够
			// 导致分裂。
			const uint64_t file_size = total_file_size_no_lock();
			const uint64_t max_size = data_size_threshold();

			return (file_size < max_size);
		}
		else
		{
			return false;
		}
	}

	//
	// 合并紧缩
	//
	// 取最后面的几个新生成的sstable合并为一个sstable，不进行数据过
	// 滤（现在内存表紧缩时也不过滤数据了，仅在主要紧缩时过滤数据）。
	// 合并完成后，删除原sstable，加入新的sstable；该过程中，不能有
	// 新的SSTable加入，否则下次取到的就是合并后的sstable且顺序不一致了。
	//
	// 但合并过程中不持有锁，可能会发生次要紧缩来加入新的sstable；
	// 因此，在合并完成后，删除sstable的时候特别注意了范围，且新加
	// 入的sstable也特别注意了插入位置。
	//
	// sstable数通过合并紧缩计数来计算获得，该计算方式应该能保证平
	// 均情况下，合并的文件总是刚次要紧缩生成的。
	//
	// 计算方式为：
	// 设最多SSTable数为S；
	// 初始有 N 个sstable，则置合并紧缩计数 C＝N ；
	// 每次当SSTable数达到S后，取后面的 S－C 个sstable进行合并（即新生成的）；
	// 合并后，置 C=C+1 ，即这时候的大sstable数；
	// 直到 S－C＜MIN ，进行主要紧缩，并置 C=1
	//
	// 实际中，MIN就是lower_bound_of_max_num_sstable；每次合并可能
	// 没有生成新的SSTable，此时C不加一（即保持与合并后的大sstable数
	// 一致）
	//
	// 初始的N个sstable不一定都是合并后的，但没有关系，这仅在子表刚
	// 被加载时发生。
	//
	void group::merge_compact() {
		LOG(TRACE, "starting merge compact");
		LOG_ENV(uint64_t start_time = cur_micro_seconds());

		const std::size_t max_sstable_num = m_group_info.m_max_num_sstable;
		LOG_ENV(const std::size_t lower_bound = m_lower_bound_of_max_num_sstable);
		assert(max_sstable_num >= (m_merge_counter + lower_bound)); // 进入前必须检查通过
		const std::size_t sstable_num_to_merge = max_sstable_num - m_merge_counter;
		
		merge_compact_scanner scanner;
		std::vector<std::string> columns(1);
		columns[0] = "*";

		boost::shared_lock<boost::shared_mutex> slock(m_sstables_lock);

		const std::size_t sstable_num = m_sstables.size();
		const std::size_t merge_num = (std::min)(sstable_num, sstable_num_to_merge);
		LOG(INFO, merge_num << " sstables to merge, total sstable num = " << sstable_num);
		typedef std::vector<sstable::pointer>::iterator it_t;
		const std::size_t start_index = sstable_num - merge_num;
		const it_t end = m_sstables.end();
		const it_t start = end - merge_num;
		assert(start == (m_sstables.begin() + start_index));
		for(it_t pos = start; pos != end; ++pos)
		{
			kv_scan_context::pointer context(new kv_scan_context(
								 tablet_name(),
								 "", "",
								 columns,
								 NO_TIMESTAMP, // latest_ts
								 NO_TIMESTAMP, // oldest_ts
								 0, // max ttl
								 0, // version
								 (kv_scan_context::OPTION_DISABLE_FILTER_DELETED_ITSELF ||
								  kv_scan_context::OPTION_DISABLE_FILTER_DELETED),
								 1 // num row
								 ));
			kv_scanner::pointer p = (*pos)->get_scanner(context);
			if(! p)
			{
				LOG(WARN, "got null sstable scanner in sstable "
				    << (*pos)->path() << ", skip it");
			}
			else
			{
				LOG(INFO, "got scanner from sstable " << (*pos)->path());
				scanner.add(p);
			}
		}

		slock.unlock();

		sstable::pointer merge_sstable;
		if(scanner.eos())
		{
			LOG(DEBUG, "scanner is empty, can delete sstables");
			// 这里不要返回，要到后面的处理逻辑
		}
		else
		{
			const uint64_t seq_num = this->max_sequence_number() + 1;
			assert(seq_num > 1);
			LOG(INFO, "merge compacting with sequence number " << seq_num
			    << " in group " << group_name() << " of tablet " << tablet_name());
			merge_sstable = sstable::create(naming_strategy::sstable_path(m_path.string(),
										      seq_num),
							naming_strategy::compact_sstable_path(m_path.string(),
											      seq_num),
							naming_strategy::deleted_sstable_path(m_path.string(),
											      seq_num),
							m_start_row,
							m_end_row,
							seq_num,
							m_group_info.m_block_size,
							static_cast<compressor::compressor_code>(m_group_info.m_compress_method));
			if(! merge_sstable)
			{
				LOG(ERROR, "create sstable with seq_num "
				    << seq_num << " in group path " << m_path
				    << " failed, abort merging");
				return;
			}

			LOG_ENV(std::size_t kv_count = 0);
			for(; ! scanner.eos(); scanner.next())
			{
				const bool ok = merge_sstable->append(scanner.get());
				if(! ok)
				{
					LOG(ERROR, "appending record ("
					    << dump_string_char(*scanner.get().m_key->row())
					    << ", "
					    << dump_string_char(*scanner.get().m_key->column())
					    << ", "
					    << scanner.get().m_key->timestamp()->get()
					    << ") into merge sstable failed, sstable = "
					    << merge_sstable->path());
					return;
				}
				LOG_ENV(++kv_count);
			}
			LOG(DEBUG, kv_count << " records written into merge sstable "
			    << merge_sstable->path());

			const bool ok = merge_sstable->append_eof();
			if(! ok)
			{
				LOG(ERROR, "append eof failed, removing the merge sstable "
				    << merge_sstable->path());
				merge_sstable->set_deleted();
				return;
			}
		}

		// 去掉原来的被合并的sstable，加入合并后的sstable
		//Guard<ReadWriteLock::WriteLock> _guard(m_sstables_lock.write_lock());
		boost::unique_lock<boost::shared_mutex> _guard(m_sstables_lock);
		// 重新计算end和start
		const it_t nstart = m_sstables.begin() + start_index;
		const it_t nend = start + merge_num;
		for(it_t pos = nstart; pos != nend; ++pos)
			{
				LOG_ENV(const bool ok_deleted =)
					(*pos)->set_deleted();
				LOG_IF_ELSE((! ok_deleted),
					    ERROR,
					    "set_deleted failed, ignore it, sstable = "
					    << (*pos)->path(),
					    INFO,
					    "old sstable file "
					    << (*pos)->path() << " set deleted");
				// 如果这里删除旧文件出错，则会导致数据
				// 重复。但对数据影响不大，下次紧缩时，
				// 还会再次删除掉它。
				//
				// 这就要求，在读取数据的时候，要严格按
				// 照SSTable文件的新旧顺序来读，如果
				// Key完全相同，则应该取较新的文件来读！
			}
		const it_t insert_pos = m_sstables.erase(nstart, nend);
		if(merge_sstable)
		{
			LOG_IF((insert_pos != m_sstables.end()), INFO,
			       "having new sstable added during merging");
			m_sstables.insert(insert_pos, merge_sstable);
			m_max_sequence_number = (std::max)(m_max_sequence_number, merge_sstable->sequence_number());
			// 成功添加后，设置 m_merge_counter 自增。
			++ m_merge_counter;
		}
		// 此时，sstable数应该和merge_counter一致
		LOG_ENV(const std::size_t cur_sstable_num = m_sstables.size());
		_guard.unlock(); // 在此解锁
		LOG_IF((m_merge_counter != cur_sstable_num),
		       INFO, "must has new sstable added, merge_counter = "
		       << m_merge_counter << ", sstable number = " << cur_sstable_num);

		LOG(INFO, "merge compact with " << merge_num
		    << " sstables used "
		    << cur_micro_seconds() - start_time
		    << " micro seconds");
	}
	
} // namespace xbase
